package DataStreamApi.Transformation算子.分流;

import domain.WaterSensor;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;
import java.util.List;

/**
 * 1输入1输出
 */
public class Flink01_FILTER {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());


        List<WaterSensor> waterSensorList = Arrays.asList(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s2", 3L, 3),
                new WaterSensor("s3", 1L, 4),
                new WaterSensor("s1", 11L, 5),
                new WaterSensor("s1", 2L, 6));
        executionEnvironment.setParallelism(6);

        DataStreamSource<WaterSensor> waterSensorDataStreamSource = executionEnvironment.fromCollection(waterSensorList);


        SingleOutputStreamOperator<WaterSensor> filterS1 = waterSensorDataStreamSource.filter(new FilterFunction<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor value) throws Exception {
                return value.getId().equals("s1");
            }
        });

        SingleOutputStreamOperator<WaterSensor> filterS2 = waterSensorDataStreamSource.filter(new FilterFunction<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor value) throws Exception {
                return value.getId().equals("s2");
            }
        });


        filterS1.print("s1流");
        filterS2.print("s2流");
        executionEnvironment.execute();
    }
}
